package edu.purdue.cs.aggr.accumulators;

import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import edu.purdue.cs.aggr.AbstractAccumulator;
import edu.purdue.cs.aggr.Accumulator;

public class WordCountAccumulator extends AbstractAccumulator implements Accumulator<String, HashMap<String, Integer>> {

	HashMap<String, Integer> wordCounts = new HashMap<String, Integer>();
	
	public WordCountAccumulator(String sourceId)
  {
	  super(sourceId);
  }

  @Override
	public void writeExternal(ObjectOutput out) throws IOException {
    super.writeExternal(out);
		out.writeObject(wordCounts);
	}

	@Override
	public void readExternal(ObjectInput in) throws IOException,
			ClassNotFoundException {
	  super.readExternal(in);
		wordCounts = (HashMap<String, Integer>) in.readObject();
	}

	@Override
	public void initialize(String input) {
	}

	@Override
	public void addValue(String input) {
		Integer currentVal = wordCounts.get(input);
		if (currentVal==null) {
			currentVal = new Integer(0);
		}
		wordCounts.put(input, currentVal + 1);
	}

	@Override
	public HashMap<String, Integer> getAccumulation() {
		return wordCounts;
	}

	@Override
	public void addAccumulators(
			List<Accumulator<String, HashMap<String, Integer>>> accumulators) {
		HashMap<String, Integer> aggregate = new HashMap<String,Integer>();
		accumulators.add(this);
		
		for (Accumulator<String, HashMap<String, Integer>> acc : accumulators) {
			Map<String, Integer> wordMap = acc.getAccumulation();
			for (String word : wordMap.keySet()) {
				Integer count = new Integer(0);
				for(Accumulator<String, HashMap<String, Integer>> inneracc : accumulators) {
					if(inneracc.getAccumulation().containsKey(word)) {
						count += inneracc.getAccumulation().get(word);
						inneracc.getAccumulation().remove(word);
					}
				}
				aggregate.put(word, count);
			}
		}
		
		wordCounts = aggregate;
	}
}
